package com.visec.hdy.log;

import com.visec.hdy.hadoop.analysis.TimeAnalysisImpl;
import com.visec.hdy.hadoop.context.LogWriteContextImpl;
import com.visec.hdy.hadoop.context.TimeLogReadContextImpl;
import com.visec.hdy.hadoop.fsb.HdfsFSB;
import com.visec.hdy.hadoop.inquiry.TimeLogInquiryImpl;
import com.visec.hdy.hadoop.standard.Analysis;
import com.visec.hdy.hadoop.standard.FSB;
import com.visec.hdy.hadoop.standard.Inquiry;
import com.visec.hdy.hadoop.standard.Storage;
import com.visec.hdy.hadoop.storage.HdfsLogStorageImpl;
import com.visec.hdy.hadoop.storage.TimeCacheStorageProxy;

/**
 * hdfs日志存储接口
 * 
 * @author hdy
 *
 * @param <V>
 */
public class HdfsLog<V> implements Log<V> {
	private static final long DEFAULT_FILE_SIZE = 20 * 10000l;
	private FSB fsb;
	private Storage<String, V> storage;
	private Analysis<String, V, V> analysis;
	private Inquiry<String, V> inquiry;

	public HdfsLog(String folder, String key, String uri) {
		this.fsb = new HdfsFSB(uri);
		this.storage = new TimeCacheStorageProxy<>(new HdfsLogStorageImpl<>(this.fsb));
		this.analysis = new TimeAnalysisImpl<>(key);
		this.storage.setContext(new LogWriteContextImpl<>(getAbstractPath(folder), DEFAULT_FILE_SIZE));
		this.storage.open();
		this.inquiry = new TimeLogInquiryImpl<>(fsb);
		this.inquiry.setContext(new TimeLogReadContextImpl(getAbstractPath(folder), DEFAULT_FILE_SIZE));
	}

	private String getAbstractPath(String path) {
		return path;
	}

	@Override
	public void save(V o) {
		this.storage.store(this.analysis.analyze(o));
	}

	@Override
	public Find<V> find(long start, long end) {
		this.inquiry.query(String.valueOf(start), String.valueOf(end));
		return new Find<>(inquiry);
	}

	public void setAnalysis(Analysis<String, V, V> analysis) {
		this.analysis = analysis;
	}

	@Override
	public void close() {
		this.inquiry.clear();
		this.storage.close();
		this.fsb.close();
	}
}
